package com.iot08.api.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.iot08.api.service.supplement.TCPService;
import com.iot08.common.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * @author weichao 1299101858@qq.com
 * @version 1.0
 * @date 2021-09-14 18:04
 */
@Slf4j
public class ConsumerTask implements Runnable {


    private ConsumerRecord<String, String> taskRecord;

    //
    public ConsumerTask(ConsumerRecord<String, String> record) {
        this.taskRecord = record;
    }
    @Override
    public void run() {
        if(taskRecord != null){
            String imei = taskRecord.key();
            String value = taskRecord.value().replace("\r", "").replace("\n", "")
                    .replace("\ufeff", "");
            log.info("topic={}. key={}. value={}. partition={}", taskRecord.topic(), imei, value, taskRecord.partition());
            try{
                JSONObject data = JSON.parseObject(value);
                if(data != null && StringUtil.isNotBlank(imei)){
                    new TCPService().cmdTempFunc(imei, data);
                }
            } catch (Exception e){
                log.error(String.format("ERROR = %s imei = %s. taskRecord = %s", e.getMessage(), imei, taskRecord),e);
            }
        } else {
            log.error("listen taskRecord is null.");
        }
    }
}
